package user.service.Impl;

import com.alibaba.fastjson.JSON;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.http.client.utils.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import user.contstant.Constants;
import user.contstant.PassStatus;
import user.dao.MerchantsDao;
import user.entity.Merchants;
import user.mapper.PassRowMapper;
import user.service.IUserPassService;
import user.vo.Pass;
import user.vo.PassInfo;
import user.vo.PassTemplate;
import user.vo.Response;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 用户优惠券相关功能实现
 */
@Slf4j
@Service
public class UserPassServiceImpl implements IUserPassService {


     //HBase客户端
     private final HbaseTemplate hbaseTemplate;
     //Merchants Dao
     private final MerchantsDao merchantsDao;
     @Autowired
     public UserPassServiceImpl(HbaseTemplate hbaseTemplate, MerchantsDao merchantsDao) {
        this.hbaseTemplate = hbaseTemplate;
         this.merchantsDao = merchantsDao;
     }

    /**
     * 获取用户个人优惠券信息，即我的优惠券功能实现
     * @param userId 用户id
     * @return  {@link Response}
     */
    @Override
    public Response getUserPassInfo(Long userId) throws Exception {

        return getPassInfoByStatus(userId,PassStatus.UNUSED);
    }

    /**
     * 获取用户已经消费了的优惠券，即已使用优惠券功能实现
     * @param userId 用户id
     * @return  {@link Response}
     * @throws Exception
     */
    @Override
    public Response getUserUsedPassInfo(Long userId) throws Exception {
        return getPassInfoByStatus(userId,PassStatus.USED);
    }

    /**
     * 获取用户所有的优惠券
     * @param userId 用户id
     * @return  {@link Response}
     * @throws Exception
     */
    @Override
    public Response getUserAllPassInfo(Long userId) throws Exception {
        return getPassInfoByStatus(userId,PassStatus.ALL);
    }

    /**
     * 用户使用优惠券
     * @param pass {@link Pass}
     * @return {@link Response}
     */
    @Override
    public Response userUsePass(Pass pass) {

        //根据userId构造行键前缀
        byte[] rowPrefix = Bytes.toBytes(new StringBuilder(
                String.valueOf(pass.getUserId())).reverse().toString());
        //设置扫描器
        Scan scan = new Scan();
        //过滤器列表
        List<Filter> filters = new ArrayList<>();
        filters.add(new PrefixFilter(rowPrefix));
        filters.add(new SingleColumnValueFilter(
                Constants.PassTable.FAMILY_I.getBytes(),
                Constants.PassTable.TEMPLATE_ID.getBytes(),
                CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes(pass.getTemplateId())
        ));
        filters.add(new SingleColumnValueFilter(
                Constants.PassTable.FAMILY_I.getBytes(),
                Constants.PassTable.CON_DATE.getBytes(),
                CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes(-1)
        ));
        scan.setFilter(new FilterList(filters));
        //查找
        List<Pass> passes = hbaseTemplate.find(Constants.PassTable.TABLE_NAME,scan,new PassRowMapper());

        //查找失败
        if(passes == null || passes.size() != 1){
            log.error("userUserPass error:{}", JSON.toJSONString(pass));
            return Response.failure("userUserPass error");
        }

        /**查询成功后修改数据*/
        byte[] FAMILY_I = Constants.PassTable.FAMILY_I.getBytes();
        byte[] CON_DATE = Constants.PassTable.CON_DATE.getBytes();

        List<Mutation> datas = new ArrayList<>();
        Put put = new Put(passes.get(0).getRowKey().getBytes());
        put.addColumn(FAMILY_I,CON_DATE,Bytes.toBytes(DateFormatUtils.ISO_DATE_FORMAT.format(new Date())));
        datas.add(put);

        hbaseTemplate.saveOrUpdate(Constants.PassTable.TABLE_NAME, (Mutation) datas);
        return Response.success();
    }

    /**
     * 根据优惠券状态获取优惠券信息
     * @param userId 用户id
     * @param passStatus {@link PassStatus}
     * @return {@link Response}
     */
    private Response getPassInfoByStatus(Long userId, PassStatus passStatus) throws Exception {

        //根据userId构造行建前缀
        byte[] rowPreFix = Bytes.toBytes(new StringBuilder(String.valueOf(userId)).reverse().toString());

        CompareFilter.CompareOp compareOp =
                passStatus == PassStatus.UNUSED ?
                CompareFilter.CompareOp.EQUAL : CompareFilter.CompareOp.NOT_EQUAL;
        Scan scan = new Scan();

        //过滤器列表
        List<Filter> list = new ArrayList<>();
        //1.行键前缀过滤器，找到特定用户的优惠券
        list.add(new PrefixFilter(rowPreFix));
        //2.基于列单元值的过滤器，找到未使用的优惠券
        if(passStatus != PassStatus.ALL)
            //单列值过滤，消费日期过滤
            list.add(new SingleColumnValueFilter(
                    Constants.PassTable.FAMILY_I.getBytes(),
                    Constants.PassTable.CON_DATE.getBytes(),
                    compareOp,Bytes.toBytes(-1)));
        scan.setFilter(new FilterList(list));

        //转换为对象
        List<Pass> passes = hbaseTemplate.find(Constants.PassTable.TABLE_NAME,scan,new PassRowMapper());
        //转换为passTemplateMap，优惠券集合
        Map<String,PassTemplate> passTemplateMap = buildPassTemplate(passes);
        //转换为MerchantsMap，商家集合
        Map<Integer,Merchants> merchantsMap = buildMerchantMap(
                new ArrayList<>(passTemplateMap.values()));
        List<PassInfo> result = new ArrayList<>();
        for(Pass pass : passes){
            /** 获取用户优惠券信息*/
            PassTemplate passTemplate = passTemplateMap.getOrDefault(
                    pass.getTemplateId(),null);
            if(passTemplate == null) {
                log.error("passTemplate null : {}",pass.getTemplateId());
                continue;
            }
            /**获取商户信息*/
            Merchants merchants = merchantsMap.getOrDefault(
                    passTemplate.getId(),null);
            if(merchants == null){
                log.error("merchant null : {}",passTemplate.getId());
                continue;
            }
            result.add(new PassInfo(pass,passTemplate,merchants));
        }
        return new Response(result);
    }
    /**
     * 通过获取的Pass对象构造Map
     * @param passes {@link Pass}
     * @return Map {@link PassTemplate}
     * @throws Exception
     */
    private Map<String, PassTemplate> buildPassTemplate(List<Pass> passes) throws Exception{

        //定义时间格式
        String[] patterns = new String[]{"yyyy-MM-dd"};

        //列族信息
        byte[] FAMILY_B = Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B);
        byte[] ID = Bytes.toBytes(Constants.PassTemplateTable.ID);
        byte[] TITLE = Bytes.toBytes(Constants.PassTemplateTable.TITLE);
        byte[] SUMMARY = Bytes.toBytes(Constants.PassTemplateTable.SUMMARY);
        byte[] DESC = Bytes.toBytes(Constants.PassTemplateTable.DESC);
        byte[] HAS_TOKEN = Bytes.toBytes(Constants.PassTemplateTable.HAS_TOKEN);
        byte[] BACKGROUND = Bytes.toBytes(Constants.PassTemplateTable.BACKGROUND);


        byte[] FAMILY_C = Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C);
        byte[] LIMIT = Bytes.toBytes(Constants.PassTemplateTable.LIMIT);
        byte[] START = Bytes.toBytes(Constants.PassTemplateTable.START);
        byte[] END = Bytes.toBytes(Constants.PassTemplateTable.END);

        /** 获取template RowKey集合*/
        List<String> templateIds = passes.stream().map(
                Pass::getTemplateId
        ).collect(Collectors.toList());

        /** 通过RowKey 获取Get 对象*/
        List<Get> templateGets = new ArrayList<>(templateIds.size());
        templateIds.forEach(t -> templateGets.add(new Get(Bytes.toBytes(t))));

        //建立连接获取表实例获取Result结果集
        Result[] templateResults = hbaseTemplate.getConnection()
                .getTable(TableName.valueOf(Constants.PassTemplateTable.TABLE_NAME))
                .get(templateGets);

        //构造 PassTemplateId -> PassTemplate Object 的 Map，用于构造PassInfo
        Map<String, PassTemplate> templateMap = new HashMap<>();
        for (Result item : templateResults){
            //构造passTemplate 对象
            PassTemplate passTemplate = new PassTemplate();
            passTemplate.setId(Bytes.toInt(item.getValue(FAMILY_B,ID)));
            passTemplate.setTitle(Bytes.toString(item.getValue(FAMILY_B,TITLE)));
            passTemplate.setSummary(Bytes.toString(item.getValue(FAMILY_B,SUMMARY)));
            passTemplate.setDesc(Bytes.toString(item.getValue(FAMILY_B,DESC)));
            passTemplate.setHasToken(Bytes.toBoolean(item.getValue(FAMILY_B,HAS_TOKEN)));
            passTemplate.setBackground(Bytes.toInt(item.getValue(FAMILY_B,BACKGROUND)));
            passTemplate.setLimit(Bytes.toLong(item.getValue(FAMILY_C,LIMIT)));
            passTemplate.setStart(DateUtils.parseDate(
                    Bytes.toString(item.getValue(FAMILY_C,START)),patterns));
            passTemplate.setEnd(DateUtils.parseDate(
                    Bytes.toString(item.getValue(FAMILY_C,END)),patterns));

            templateMap.put(Bytes.toString(item.getRow()),passTemplate);
        }
        return templateMap;
    }

    /**
     * 通过获取的passTemplate 对象构造 Merchants Map
     * @param   passTemplates {@link PassTemplate}
     * @return Map {@link Merchants}
     */
    private Map<Integer, Merchants> buildMerchantMap(List<PassTemplate> passTemplates){

        Map<Integer,Merchants> merchantsMap = new HashMap<>();
        //获取passTemplates包含的商户id
        List<Integer> merchantIds = passTemplates.stream().map(
                PassTemplate::getId
        ).collect(Collectors.toList());

        //根据merchantId查找merchant
        List<Merchants> merchants = merchantsDao.findByIdIn(merchantIds);

        merchants.forEach(m -> merchantsMap.put(m.getId(),m));

         return merchantsMap;
    }
}
